/*
 * Copyright (c)  2020,  DataPipeline
 * All rights reserved.
 *
 * Id:SimpleProducer.java   2020-07-26 22:34 wanglijun
 */
package cn.datacare.kafka.stream.wordcount;

import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.*;
import java.net.URISyntaxException;
import java.util.Properties;

/**
 * <p>
 * Title:
 * </p>
 * <p>
 * Description:
 * </p>
 * <p>
 * Copyright: Copyright (c) 2020
 * </p>
 * <p>
 * Company/Department: DataPipeline
 * </p>
 *
 * @author wanglijun
 * @version 1.0
 * @since 2020-07-26 22:34
 */
public class SimpleProducer {

    public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {

        // Assign topicName to string variable
        String topicName = "TextLinesTopic2";

        Producer<String, String> producer = new KafkaProducer<String, String>(getProps());

        // read a txt file , send it line by line
        File file = new File(SimpleProducer.class.getResource("/").getPath()+"wordcount.txt");
        BufferedReader reader = new BufferedReader(new FileReader(file));
        String tempString = null;
        while ((tempString = reader.readLine()) != null) {
            producer.send(new ProducerRecord<>(topicName, tempString));
            Thread.sleep(1000);
        }
        reader.close();

        /*
         * for (int i = 0; i < 10; i++) { producer.send(new ProducerRecord<String, String>(topicName, Integer.toString(i), Integer.toString(i))); }
         */
        System.out.println("Message sent successfully");
        producer.close();
    }

    public static Properties getProps(){
        // create instance for properties to access producer configs
        Properties props = new Properties();

        // Assign localhost id, 参考http://kafka.apache.org/documentation/#producerapi
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // Set acknowledgements for producer requests.
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        // If the request fails, the producer can automatically retry,
        props.put(ProducerConfig.RETRIES_CONFIG, 0);

        // Specify buffer size in config
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

        // Reduce the no of requests less than 0
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);

        // The buffer.memory controls the total amount of memory available to the
        // producer for buffering.
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        return props;

    }
}
